package com.wuwangfu.checkepoint;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
import org.apache.flink.util.Collector;

import java.util.Properties;

/**
 * @Description：数据一致性实现：为状态添加 checkpoint（检查点）
 * @Author：jcshen
 * @Date：2023-07-02
 *
 * https://nightlies.apache.org/flink/flink-docs-release-1.17/zh/docs/dev/datastream/fault-tolerance/checkpointing/
 *
 * 当前程序能不能容错（保证数据一致性）
 * 如果可以保证数据的一致性，是使用ExactlyOnce还是AtLeastOnce，使用的是AtLeastOnce
 *
 * KafkaSource：可以记录偏移量，可以将偏移量保存到状态中（OperatorState）
 * KeyBy后调用sum：sum有状态（ValueState）
 * RedisSink：使用hset方法可以将数据覆盖（幂等性）
 *
 * 重启job从checkpoint恢复
 *
 * savepoint
 *  bin/flink stop -p hdfs://node-01:9000/flink/savepoint jodID
 *
 *  bing/flink -p 4 -c com.wuwangfu.checkepoint.KafkaToRedis \
 *  -s hdfs://node-01:9000/flink/savepoint/savepoint-3d4f5g-22791fc01408/_metadata \
 *  /data/flink/job/flink-policy-analysis.jar \
 *  --topic policy-excel --groupId g01 --redisHost node-01 --redisPwd 123456 \
 *  --fsBackend hdfs://node-01:9000/flink/checkpoint/
 *
 *
 *
 */
public class KafkaToRedis {
    public static void main(String[] args) throws Exception {
        //参数传递
        /**
         * --topic wc
         * --groupId wc
         */
        ParameterTool params = ParameterTool.fromArgs(args);
        System.setProperty("HADOOP_USER_NAME","hadoop");


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置checkpoint
        env.enableCheckpointing(30000);
        //设置状态存储后端：向hdfs写checkpoint
        env.setStateBackend(new FsStateBackend("hdfs://node-01.com:9000/flink/checkpoint"));
        //如果手动cancel job后，不删除job的checkpoint数据
        env.getCheckpointConfig().enableExternalizedCheckpoints(
                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        /*添加KafkaSource*/
        Properties propers = new Properties();
        //设置Kafka的地址和端口
        propers.setProperty("bootstrap.servers","node-01.com:9200");
        //读取偏移量策略：如果没有偏移量策略，就从头读，如果记录过偏移量就，接着读
        propers.setProperty("auto.offset.reset","earliest");
        //设置消费组ID
//        propers.setProperty("group.id","g1");
        propers.setProperty("group.id",params.get("groupId"));
        //没有开启checkpoint，不让flink提交偏移量的消费者定期自动提交偏移量
        //不让flink的消费（source对它的subtask）自动提交偏移量
        propers.setProperty("enable.auto.commit","false");
        //创建FlinkKafkaConsumer并传入相关参数
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(
                params.getRequired("topic"),//topic名称
                new SimpleStringSchema(),//读取文件的反序列化Schema
                propers//传入Kafka的参数
        );
        /**
         * 查询__consumer_offsets topic 所有内容
         * bin/kafka-console-consumer.sh --topic __consumer_offsets \
         * --formatter "kafka.coordinator.group.GroupMetadataManager$OffsetsMessageFormatter" \
         * --consumer.config config/consumer.properties --from-beginning \
         * | grep  wc
         */
        /*设置在checkpoint时不将偏移量保存到Kafka特殊的topic中*/
        kafkaConsumer.setCommitOffsetsOnCheckpoints(false);

        /*使用addSource添加kafakConsumer*/
        DataStreamSource<String> lines = env.addSource(kafkaConsumer);
        //
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = lines.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String line, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] words = line.split("");
                for (String word : words) {
                    out.collect(Tuple2.of(word, 1));
                }
            }
        });
        //分组
        KeyedStream<Tuple2<String, Integer>, String> keyed = wordAndOne.keyBy(t->t.f0);
        //聚合
        SingleOutputStreamOperator<Tuple2<String, Integer>> summed = keyed.sum(1);
        //连接池
        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder()
                .setHost("127.0.0.1")
                .setPassword("")
                .setDatabase(0)
                .build();
        //写入redis
        summed.addSink(new RedisSink<Tuple2<String, Integer>>(conf, new RedisWordCountMapper()));

        env.execute();
    }

    private static class RedisWordCountMapper implements RedisMapper<Tuple2<String, Integer>> {
        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.HSET,"WORD_COUNT");
        }

        @Override
        public String getKeyFromData(Tuple2<String, Integer> data) {
            return data.f0;
        }

        @Override
        public String getValueFromData(Tuple2<String, Integer> data) {
            return data.f1.toString();
        }
    }

}
